package com.ffcs.crmd.tsp.task.netty.processor;

import java.util.Date;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import org.slf4j.Logger;

import com.ffcs.crmd.tsp.api.dto.job.TspJobDetailShell;
import com.ffcs.crmd.tsp.api.dto.netty.RemotingCommand;
import com.ffcs.crmd.tsp.api.dto.task.TspJobExecutionException;
import com.ffcs.crmd.tsp.api.dto.task.TspJobInstance;
import com.ffcs.crmd.tsp.api.netty.INettyInvokeCallback;
import com.ffcs.crmd.tsp.api.netty.INettyProcessor;
import com.ffcs.crmd.tsp.api.netty.INettyRemotingClient;
import com.ffcs.crmd.tsp.api.task.job.IJob;
import com.ffcs.crmd.tsp.api.task.job.IJobBatch;
import com.ffcs.crmd.tsp.api.task.job.IJobExecutionContext;
import com.ffcs.crmd.tsp.common.constant.LoggerName;
import com.ffcs.crmd.tsp.common.constant.TspAppConstant;
import com.ffcs.crmd.tsp.common.exception.TspRemotingException;
import com.ffcs.crmd.tsp.common.netty.RequestCode;
import com.ffcs.crmd.tsp.common.netty.ResponseCode;
import com.ffcs.crmd.tsp.core.util.BeanUtils;
import com.ffcs.crmd.tsp.core.util.RemotingUtil;
import com.ffcs.crmd.tsp.core.util.XmlUtil;
import com.ffcs.crmd.tsp.task.container.ThreadPoolContainer;
import com.ffcs.crmd.tsp.task.job.BusinessProcessor;
import com.ffcs.crmd.tsp.task.log.TaskLogger;
import com.ffcs.crmd.tsp.task.threadpool.JobProcessBusinessThreadPool;
import com.ffcs.crmd.tsp.task.threadpool.JobProcessThreadPool;
import com.ffcs.crmd.tsp.task.threadpool.api.IJobProcessRunnable;

import io.netty.channel.ChannelHandlerContext;

/**
 * 
 * 功能说明:有状态子任务处理者
 *
 * @author ZHONGFUHUA
 * 
 * @since 1.0.0-SNAPSHOT
 *
 */
public class JobDetailStateProcessor<T> implements INettyProcessor {
    
    /**
     * 日志
     */
    private static final Logger                  LOG                          = TaskLogger.getLogger(LoggerName.TaskLoggerName);
    
    /**
     * 任务处理线程池
     */
    private JobProcessThreadPool                 jobProcessThreadPool         = ThreadPoolContainer.getInstance().getJobProcessThreadPool();
    
    /**
     * 任务批量业务处理线程池
     */
    private JobProcessBusinessThreadPool         jobProcessBusinessThreadPool = ThreadPoolContainer.getInstance()
        .getJobProcessBusinessThreadPool();
    
    /**
     * 
     */
    private final Map<TspJobDetailShell, String> returnFailShellCache         = new ConcurrentHashMap<TspJobDetailShell, String>();
    
    /**
     * 执行端通信层
     */
    private INettyRemotingClient                 nettyRemotingClient;
    
    /**
     * 功能说明:
     * 
     * @param nettyRemotingClient
     */
    public JobDetailStateProcessor(INettyRemotingClient nettyRemotingClient) {
        this.nettyRemotingClient = nettyRemotingClient;
    }
    
    @Override
    public RemotingCommand processRequest(final ChannelHandlerContext ctx, final RemotingCommand cmd) {
        TspJobDetailShell tspJobDetailShell = XmlUtil.fromXml(new String(cmd.getBody()), TspJobDetailShell.class);
        LOG.debug("TSP-TASK-JOB: recive a request to run stateful job detail[{}] id=[{}] opqque=[{}].",
            tspJobDetailShell.getContext().getTspJobInstance().getJobKey(), tspJobDetailShell.getId(), cmd.getOpaque());
        
        joinJobDetailToRun(ctx, cmd, tspJobDetailShell);
        
        RemotingCommand response = RemotingCommand.createResponseCommand(ResponseCode.SUCCESS);
        response.setBody((String.valueOf(new Date().getTime())).getBytes());
        return response;
    }
    
    /**
     * 
     * 功能说明:把子任务加入线程池中执行
     * 
     * @param ctx netty通道上下文
     * @param cmd 请求报文
     * @param tspJobDetailShell 子任务运行内核通信体
     */
    @SuppressWarnings({"unchecked", "deprecation" })
    private void joinJobDetailToRun(final ChannelHandlerContext ctx, final RemotingCommand cmd, final TspJobDetailShell tspJobDetailShell) {
        final IJobExecutionContext<T> context = (IJobExecutionContext<T>) tspJobDetailShell.getContext();
        final TspJobInstance tspJobDetail = context.getTspJobInstance();
        
        final com.ffcs.crmd.tsp.core.job.TspJobExecutionContext<Object> tspJobExecutionContextOld = new com.ffcs.crmd.tsp.core.job.TspJobExecutionContext<Object>();
        tspJobExecutionContextOld.setFireInstanceId(context.getFireInstanceId());
        tspJobExecutionContextOld.setFireTime(context.getFireTime());
        tspJobExecutionContextOld.setJobRunTime(context.getJobRunTime());
        tspJobExecutionContextOld.setNextFireTime(context.getNextFireTime());
        tspJobExecutionContextOld.setPreviousFireTime(context.getPreviousFireTime());
        tspJobExecutionContextOld.setRecovering(context.isRecovering());
        tspJobExecutionContextOld.setRefireCount(context.getRefireCount());
        tspJobExecutionContextOld.setScheduledFireTime(context.getScheduledFireTime());
        
        if (context.getTspJobDataMap() != null) {
            tspJobExecutionContextOld.getTspJobDataMap().putAll(context.getTspJobDataMap().getMap());
            if(context.getTspJobDataMap().get(TspAppConstant.TSP_PARAM_SLICE_GROUP) != null){
                String slices = String.valueOf(context.getTspJobDataMap().get(TspAppConstant.TSP_PARAM_SLICE_GROUP));
                tspJobExecutionContextOld.setSlices(slices.split(","));
            }else if(context.getTspJobDataMap().get(TspAppConstant.TSP_PARAM_SLICES) != null){
                String slices = String.valueOf(context.getTspJobDataMap().get(TspAppConstant.TSP_PARAM_SLICES));
                tspJobExecutionContextOld.setSlices(slices.split(","));
            }
        }
        
        com.ffcs.crmd.tsp.common.job.entity.TspJobDetail tspJobDetailOld = new com.ffcs.crmd.tsp.common.job.entity.TspJobDetail();
        try {
            BeanUtils.applyIf(tspJobDetailOld, tspJobDetail);
            tspJobDetailOld.setGroup(tspJobDetail.getJobName());
        } catch (Exception e) {
            LOG.error("TSP-TASK-JOB: copy new tspJobDetail entity to tspJobDetailOld exception:", e);
        }
        tspJobExecutionContextOld.setTspJobDetail(tspJobDetailOld);
        
        jobProcessThreadPool.submit(tspJobDetailShell, tspJobDetailShell.getPriority(), new IJobProcessRunnable() {
            
            /**
             * 上下文
             */
            private IJobExecutionContext<T>                                contexts                 = context;
            
            /**
             * 上下文
             */
            private com.ffcs.crmd.tsp.task.job.api.IJobExecutionContext<T> contextsOld              = (com.ffcs.crmd.tsp.task.job.api.IJobExecutionContext<T>) tspJobExecutionContextOld;
            
            /**
             * 子任务
             */
            private TspJobInstance                                         tspJobDetailN            = tspJobDetail;
            
            /**
             * 子任务内核通信体
             */
            private TspJobDetailShell                                      tspJobDetailShellN       = tspJobDetailShell;
            
            /**
             * 任务子线程锁(用于移除jobThreadMap)
             */
            private Lock                                                   completeLock             = new ReentrantLock();
            
            /**
             * 任务等待响应锁
             */
            private CountDownLatch                                         returnWaitting           = new CountDownLatch(1);
            
            /**
             * 任务线程缓存标记线程是否都已执行完毕
             */
            private Map<BusinessProcessor<T>, T>                           businessProcessThreadMap = new ConcurrentHashMap<BusinessProcessor<T>, T>();
            
            @Override
            public void run() {
                try {
                    Class<?> jobClass = Class.forName(tspJobDetailN.getJobClass());
                    Object jobInstanceObject = jobClass.newInstance();
                    if (jobInstanceObject instanceof IJob) {
                        executeIJob((IJob<T>) jobInstanceObject);
                    } else if (jobInstanceObject instanceof com.ffcs.crmd.tsp.task.job.api.IJob) {
                        executeIJob((com.ffcs.crmd.tsp.task.job.api.IJob<T>) jobInstanceObject);
                    } else if (jobInstanceObject instanceof com.ffcs.crmd.tsp.task.job.api.IJobBatch) {
                        executeIJobBatch((com.ffcs.crmd.tsp.task.job.api.IJobBatch<T>) jobInstanceObject);
                    } else {
                        executeIJobBatch((IJobBatch<T>) jobInstanceObject);
                    }
                } catch (Exception e) {
                    LOG.error("TSP-TASK-JOB: complete a exception stateful job detail[{}] id=[{}] opaque=[{}]: ", tspJobDetailN.getJobKey(),
                        tspJobDetailShell.getId(), cmd.getOpaque(), e);
                    
                    TspJobExecutionException tspJobExecutionException = null;
                    if (e instanceof TspJobExecutionException) {
                        tspJobExecutionException = (TspJobExecutionException) e;
                    } else {
                        tspJobExecutionException = new TspJobExecutionException(e);
                    }
                    tspJobDetailShell.setTspJobExecutionException(tspJobExecutionException);
                }
                
                sendCompleteJobDetail(tspJobDetailShell, RemotingUtil.parseChannelRemote(ctx.channel()));
                LOG.debug("TSP-TASK-JOB: complete return a stateful job detail[{}] id=[{}] opaque=[{}]", tspJobDetailN.getJobKey(),
                    tspJobDetailShell.getId(), cmd.getOpaque());
            }
            
            /**
             * 
             * 功能说明:执行任务
             * 
             * @param jobDetailInstance 子任务实例
             * @throws TspJobExecutionException
             */
            private void executeIJob(IJob<T> jobDetailInstance) throws TspJobExecutionException {
                jobDetailInstance.execute(contexts);
            }
            
            /**
             * 
             * 功能说明:执行任务
             * 
             * @param jobDetailInstance 子任务实例
             * @throws TspJobExecutionException
             */
            private void executeIJob(com.ffcs.crmd.tsp.task.job.api.IJob<T> jobDetailInstance) throws TspJobExecutionException {
                try {
                    jobDetailInstance.execute(contextsOld);
                } catch (com.ffcs.crmd.tsp.common.exception.TspJobExecutionException e) {
                    throw new TspJobExecutionException(e);
                }
            }
            
            /**
             * 
             * 功能说明:执行批量任务
             * 
             * @param jobDetailBatchInstance 批量子任务实例
             * @throws TspJobExecutionException
             */
            private void executeIJobBatch(IJobBatch<T> jobDetailBatchInstance) throws TspJobExecutionException {
                jobDetailBatchInstance.execute(contexts);
                if (!contextsOld.getParam().isEmpty()) {
                    createBatchBusinessProcessThread(contextsOld.getParam(), jobDetailBatchInstance, tspJobDetailShell.getPriority());
                } else {
                    LOG.warn(
                        "TSP-TASK-JOB: run a stateful job detail[{}] id=[{}] opaque=[{}] success but batch params is null and java class [{}] ",
                        tspJobDetailN.getJobKey(), tspJobDetailShellN.getId(), cmd.getOpaque(), jobDetailBatchInstance.getClass().getName());
                }
            }
            
            /**
             * 
             * 功能说明:执行批量任务
             * 
             * @param jobDetailBatchInstance 批量子任务实例
             * @throws TspJobExecutionException
             */
            private void executeIJobBatch(com.ffcs.crmd.tsp.task.job.api.IJobBatch<T> jobDetailBatchInstance) throws TspJobExecutionException {
                try {
                    jobDetailBatchInstance.execute(contextsOld);
                    if (!contexts.getBussinessDatas().isEmpty()) {
                        createBatchBusinessProcessThread(contexts.getBussinessDatas(), jobDetailBatchInstance, tspJobDetailShell.getPriority());
                    } else {
                        LOG.warn(
                            "TSP-TASK-JOB: run a stateful job detail[{}] id=[{}] opaque=[{}] success but batch params is null and java class [{}] ",
                            tspJobDetailN.getJobKey(), tspJobDetailShellN.getId(), cmd.getOpaque(), jobDetailBatchInstance.getClass().getName());
                    }
                } catch (com.ffcs.crmd.tsp.common.exception.TspJobExecutionException e) {
                    throw new TspJobExecutionException(e);
                }
            }
            
            /**
             * 
             * 功能说明:创建批量业务数据处理线程
             * 
             * @param businessDataList 业务数据
             * @param jobDetailInstance 子任务实例
             * @param priority 优先级
             */
            private void createBatchBusinessProcessThread(List<T> businessDataList, IJobBatch<T> jobDetailInstance, int priority) {
                for (T businessData : businessDataList) {
                    BusinessProcessor<T> businessProcessor = new BusinessProcessor<T>(jobDetailInstance, businessData, businessProcessThreadMap,
                        completeLock, returnWaitting);
                    businessProcessThreadMap.put(businessProcessor, businessData);
                }
                
                Iterator<Entry<BusinessProcessor<T>, T>> it = businessProcessThreadMap.entrySet().iterator();
                while (it.hasNext()) {
                    Entry<BusinessProcessor<T>, T> entry = it.next();
                    BusinessProcessor<T> businessProcessor = entry.getKey();
                    T businessData = entry.getValue();
                    jobProcessBusinessThreadPool.submit(businessData, priority, businessProcessor);
                }
                
                try {
                    returnWaitting.await();
                } catch (InterruptedException e) {
                    LOG.error(
                        "TSP-TASK-JOB: stateful job detail[{}] id=[{}] opaque=[{}] process business datas is interrupted when waitting return.",
                        tspJobDetailN.getJobKey(), tspJobDetailShellN.getId(), cmd.getOpaque());
                }
            }
            
            /**
             * 
             * 功能说明:创建批量业务数据处理线程
             * 
             * @param businessDataList 业务数据
             * @param jobDetailInstance 子任务实例
             * @param priority 优先级
             */
            private void createBatchBusinessProcessThread(List<T> businessDataList,
                com.ffcs.crmd.tsp.task.job.api.IJobBatch<T> jobDetailInstance, int priority) {
                for (T businessData : businessDataList) {
                    BusinessProcessor<T> businessProcessor = new BusinessProcessor<T>(jobDetailInstance, businessData, businessProcessThreadMap,
                        completeLock, returnWaitting);
                    businessProcessThreadMap.put(businessProcessor, businessData);
                }
                
                Iterator<Entry<BusinessProcessor<T>, T>> it = businessProcessThreadMap.entrySet().iterator();
                while (it.hasNext()) {
                    Entry<BusinessProcessor<T>, T> entry = it.next();
                    BusinessProcessor<T> businessProcessor = entry.getKey();
                    T businessData = entry.getValue();
                    jobProcessBusinessThreadPool.submit(businessData, priority, businessProcessor);
                }
                
                try {
                    returnWaitting.await();
                } catch (InterruptedException e) {
                    LOG.error(
                        "TSP-TASK-JOB: stateful job detail[{}] id=[{}] opaque=[{}] process business datas is interrupted when waitting return.",
                        tspJobDetailN.getJobKey(), tspJobDetailShellN.getId(), cmd.getOpaque());
                }
            }
            
            @Override
            public TspJobDetailShell getTspJobDetailShell() {
                return tspJobDetailShellN;
            }
            
            @Override
            public Map<BusinessProcessor<T>, T> getBusinessProcessThreadMap() {
                return businessProcessThreadMap;
            }
            
            @Override
            public CountDownLatch getReturnWaitting() {
                return returnWaitting;
            }
            
        });
    }
    
    public void retryReturnCompleteJobDetail() {
        Iterator<Entry<TspJobDetailShell, String>> it = returnFailShellCache.entrySet().iterator();
        while (it.hasNext()) {
            Entry<TspJobDetailShell, String> entry = it.next();
            TspJobDetailShell tspJobDetailShell = entry.getKey();
            String addr = entry.getValue();
            it.remove();
            sendCompleteJobDetail(tspJobDetailShell, addr);
        }
    }
    
    private void sendCompleteJobDetail(final TspJobDetailShell tspJobDetailShell, final String addr) {
        RemotingCommand requestRemotingCommand = RemotingCommand.createRequestCommand(RequestCode.COMPLETE_JOB_DETAIL);
        requestRemotingCommand.setBody(XmlUtil.toXml(tspJobDetailShell).getBytes());
        try {
            this.nettyRemotingClient.invokeOneway(addr, requestRemotingCommand, new INettyInvokeCallback() {
                @Override
                public void operationComplete(RemotingCommand responseCommand) {
                    returnFailShellCache.put(tspJobDetailShell, addr);
                }
            });
        } catch (TspRemotingException e) {
            returnFailShellCache.put(tspJobDetailShell, addr);
            LOG.error("TSP-TASK-JOB: complete return a stateful job detail[{}] id=[{}] exception:",
                tspJobDetailShell.getContext().getTspJobInstance().getJobKey(), tspJobDetailShell.getId(), e);
        }
    }
}
